
/**
 * @file:       thread_pool.hpp
 * @author:     WubinWang
 * @contact:    wubinstu@163.com
 * @date:       2025-01-01
 * @license:    MIT License
 *
 * Copyright (c) 2025 WubinWang
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 *
 * DISCLAIMER:
 * 1. User acknowledges all application risks and consequences arising from usage
 *    of this software are exclusively borne by the user.
 * 2. The original author shall not be held liable for any direct, indirect,
 *    incidental, special, exemplary or consequential damages.
 * 3. No warranty of fitness for particular purpose or non-infringement is provided. */


#ifndef __THREAD_POOL_HPP_
#define __THREAD_POOL_HPP_

#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <ctime>
#include <functional>
#include <iostream>
#include <mutex>
#include <queue>
#include <set>
#include <thread>
#include <vector>

#include "thread_time.hpp"

/**
 * @class Thread_Pool_ct
 * @author WubinWang
 * @date 2025/01/30
 *
 * @brief The `Thread_Pool_ct` class represents a thread pool that manages a collection of worker threads
 *        to execute tasks concurrently. The thread pool dynamically adjusts the number of active threads
 *        based on the workload, ensuring efficient resource utilization and task execution.
 *
 * The thread pool provides the following features:
 * - Dynamic thread management: The pool can increase or decrease the number of worker threads based on
 *   the current workload and predefined thresholds.
 * - Task prioritization: Tasks can be added with a priority, and the pool will execute higher-priority
 *   tasks first.
 * - Task queue management: The pool maintains a queue of tasks, ensuring that tasks are executed in the
 *   order of their priority.
 * - Thread safety: The pool uses mutexes and condition variables to ensure thread-safe access to shared
 *   resources.
 * - Graceful shutdown: The pool can be shut down gracefully, ensuring that all pending tasks are completed
 *   before the threads are terminated.
 *
 * The thread pool is initialized with a minimum and maximum number of threads, a step size for adjusting
 * the number of threads, and a maximum queue size for tasks. The pool starts with the minimum number of
 * threads and dynamically adjusts the number of threads based on the workload.
 *
 * Example usage:
 * @example:
 * Thread_Pool_ct pool {                                                                       // Creat a thread pool
 *     Thread_Pool_ct::Pool_Config_st {
 *         ._queue_max      = 200,
 *         ._thread_max     = 8,
 *         ._thread_min     = 2,
 *         ._thread_step    = 1,
 *         ._auto_threshold = false,
 *     }
 * };
 *
 * pool.start ();                                                                              // Start the pool
 *
 * pool.task_add (                                                                             // Build and add the task
 *     Thread_Pool_ct::Task_Builder_st {}
 *         .id (123)
 *         .priority (456)
 *         .name ("Hello")
 *         .entry (std::bind (func, ...))
 *         .callback (callback)
 *         .build ()
 * );
 *
 * pool.stop ();                                                                               // Stop the pool gracefully
 *
 * @note The thread pool is designed to be thread-safe, but care should be taken when accessing shared
 *       resources outside the pool's control.
 */
class Thread_Pool_ct {

  public:

    using Time_ANY_t      = std::int64_t;

    using Task_ID_t       = std::uint64_t;
    using Task_Priority_t = std::uint64_t;
    using Task_Name_t     = std::string;
    using Task_Entry_t    = std::function<void *()>;
    using Task_Callback_t = std::function<void (void * _returned)>;

    using Task_State_t    = enum class Task_State_t : std::uint8_t {
        ignorable         = 0,  // 未初始化的无效任务, 或者已经完成的任务
        newlyBuild        = 1,  // 新建的有效任务, 等待处理
        onProcessing      = 2,  // 正在处理
        isFinished        = 3,  // 完成
        exceptionMain     = 4,  // 任务抛异常了
        exceptionCallback = 5,  // 回调抛异常了
    };

    using Task_Item_st = struct Task_Item_st {

      public:

        Task_Item_st (const Task_Item_st & _other) noexcept             = default;
        Task_Item_st (Task_Item_st && _other) noexcept                  = default;
        Task_Item_st & operator= (const Task_Item_st & _other) noexcept = default;
        Task_Item_st & operator= (Task_Item_st && _other) noexcept      = default;
        Task_Item_st () noexcept                                        = default;
        virtual ~Task_Item_st () noexcept                               = default;

        explicit Task_Item_st (Task_ID_t _task_id, Task_Priority_t _task_priority, Task_Name_t _task_name);

        struct Task_Comparator {
            auto operator() (const Task_Item_st & _left, const Task_Item_st & _right) const -> bool;
        };

        auto id () noexcept -> Task_ID_t &;
        auto priority () noexcept -> Task_Priority_t &;
        auto name () noexcept -> Task_Name_t &;

        auto entry () noexcept -> Task_Entry_t &;
        auto callback () noexcept -> Task_Callback_t &;

        auto time_create () noexcept -> Time_ANY_t &;
        auto time_attach () noexcept -> Time_ANY_t &;
        auto time_process () noexcept -> Time_ANY_t &;
        auto time_finish () noexcept -> Time_ANY_t &;
        auto time_cpu () noexcept -> Time_ANY_t &;
        auto time_wall () noexcept -> Time_ANY_t &;

        auto state () noexcept -> Task_State_t &;
        auto state_ignorable () noexcept -> bool;
        auto state_newlyBuild () noexcept -> bool;
        auto state_onProcessing () noexcept -> bool;
        auto state_isFinished () noexcept -> bool;

        auto thread_id () noexcept -> std::thread::id &;

        /// @brief 运行 task
        auto run () noexcept -> void;

        /// @brief 清空 task 一切信息
        auto clear () noexcept -> void;


      private:
        std::thread::id _m_process_thread {};                        // 处理这个 task 的线程 ID
        Task_ID_t       _m_task_id       = 0;                        // 任务ID, 任意给定, 线程池不会检查其唯一性, 这只是方便用户区分任务
        Task_Priority_t _m_task_priority = 0;                        // 优先级, 数值越高优先级越大
        Task_Name_t     _m_task_name     = {};                       // 任务名称, 任意给定, 方便用户识别任务
        Task_Entry_t    _m_entry         = nullptr;                  // 任务主要入口
        Task_Callback_t _m_callback      = nullptr;                  // 回调函数, 任务完成后自动调用
        Task_State_t    _m_state         = Task_State_t::ignorable;  // 任务状态
        Time_ANY_t      _m_ctime         = 0;                        // 任务创建时间
        Time_ANY_t      _m_atime         = 0;                        // 任务进队时间
        Time_ANY_t      _m_ptime         = 0;                        // 任务开始执行时间
        Time_ANY_t      _m_ftime         = 0;                        // 任务结束执行时间
        Time_ANY_t      _m_cpu_time      = 0;                        // 任务 CPU 时间 (对于实际消耗)
        Time_ANY_t      _m_wall_time     = 0;                        // 任务 WALL 时间 (对于用户)
    };

    using Task_Builder_st = struct Task_Builder_st {
        Task_Item_st _task {};

        auto id (Task_ID_t _task_id) noexcept -> Task_Builder_st &;
        auto priority (Task_Priority_t _task_priority) noexcept -> Task_Builder_st &;
        auto name (Task_Name_t _task_name) noexcept -> Task_Builder_st &;
        auto entry (Task_Entry_t _entry) noexcept -> Task_Builder_st &;
        auto callback (Task_Callback_t _callback) noexcept -> Task_Builder_st &;

        [[nodiscard]] auto build () const noexcept -> Task_Item_st;
    };

    using Pool_Config_st = struct Pool_Config_st {
        std::uint32_t _queue_max;
        std::uint32_t _thread_max;
        std::uint32_t _thread_min;
        std::uint32_t _thread_step;
        bool          _auto_threshold;
    };

    explicit Thread_Pool_ct (const Thread_Pool_ct & _other) noexcept    = default;
    explicit Thread_Pool_ct (Thread_Pool_ct && _other) noexcept         = default;
    Thread_Pool_ct & operator= (const Thread_Pool_ct & _other) noexcept = default;
    Thread_Pool_ct & operator= (Thread_Pool_ct && _other) noexcept      = default;
    explicit Thread_Pool_ct () noexcept                                 = default;
    virtual ~Thread_Pool_ct () noexcept                                 = default;

    explicit Thread_Pool_ct (Pool_Config_st _config);

    enum class Pool_State_et : std::uint8_t {
        ignorable   = 0,  // 线程池未初始化
        initialized = 1,  // 线程池完成初始化
        inRunning   = 2,  // 线程池正常运行中
        inExiting   = 3,  // 线程池正在关闭中
        isExited    = 4,  // 线程池完全关闭了
    };

    /// @brief CPU核心数
    static auto CountsOF_CoreThreads () -> std::uint32_t;

    /// @brief 获取最小线程数 (CPU核心数)
    static auto CountsOF_MinThreads () -> std::uint32_t;

    /// @brief 获取最大线程数 (基于 I/O 和 CPU 计算的优公式)
    /// @attention 时间单位需要统一
    static auto CountsOF_MaxThreads (long double _avg_wait_time, long double _avg_exec_time) -> std::uint32_t;

    /// @brief 滑动平均 (指数加权移动平均)
    /// @param _accumulated 累积值（需初始化为第一个数据或0）
    /// @param _new_data 新数据
    /// @param _alpha 权重因子 (新数据权重, 范围 (0, 1), 默认0.5)
    static auto EMAverage (std::int64_t _accumulated, std::int64_t _new_data, double _alpha = 0.5) -> std::int64_t;

    auto time_start () noexcept -> std::int64_t;
    auto time_stop () noexcept -> std::int64_t;
    auto time_accumulated_running () noexcept -> std::int64_t;
    auto time_accumulated_tasking () noexcept -> std::int64_t;
    auto time_avg_task_waiting () noexcept -> std::int64_t;
    auto time_avg_task_executing () noexcept -> std::int64_t;
    auto counts_accumulated_tasks () noexcept -> std::uint64_t;

    /// @brief 打印线程池状态信息
    auto print_state () noexcept -> void;

    /// @brief 获取线程池状态
    auto state () noexcept -> Pool_State_et;

    /// @brief 线程池初始化, 检查各个参数的正确性 (构造函数自动调用)
    auto init () noexcept -> bool;

    /// @brief 线程池启动, 工作线程和管理者线程开始运行, 工作队列开始流动
    auto start () noexcept -> bool;

    /// @brief 尝试关闭线程池, 向工作线程和管理者线程发送停止标志位, 线程会陆续退出
    /// @attention 任务队列不会显式清空 (工作线程退出时会主动将任务队列执行完毕并清空), 当然如果实在有需要则手动调用 task_clear () 函数
    auto stop () noexcept -> bool;

    /// @brief 向工作队列中添加任务
    /// @param _task 被添加的任务, 注意任务的所有权将转移给线程池, 成功调用该函数任务将失效
    /// @param _blocking (如果队列满了是否阻塞等待) 设置为 true 时阻塞等待队列不满
    auto task_add (Task_Item_st && _task, bool _blocking = true) noexcept -> bool;

    /// @brief 清空工作队列, 队列中未执行的任务会直接丢弃, 一般这个函数不会被调用
    auto task_clear () noexcept -> void;

  protected:

    /// @brief 内部函数, 不应该被用户直接访问, 这是线程池内部管理者线程的入口, 主要作用是根据负载动态调节线程池体量
    auto _thread_manager (void * _args) -> void;

    /// @brief 内部函数, 不应该被用户直接访问, 这是线程池内部工作线程的入口, 主要作用是等待和取出任务队列中的任务并执行任务
    auto _thread_worker (void * _args) -> void;

  private:
    /// std::atomic 已经重载了赋值等运算符, 对数值的"直接"操作是原子的
    std::atomic<Pool_State_et> _m_state          = Pool_State_et::ignorable;
    std::atomic<std::uint32_t> _m_thread_max     = 0;  // 最大线程数
    std::atomic<std::uint32_t> _m_thread_min     = 0;  // 最小线程池
    std::atomic<std::uint32_t> _m_thread_alive   = 0;  // 当前存活线程数
    std::atomic<std::uint32_t> _m_thread_busy    = 0;  // 当前工作中线程数
    std::atomic<std::uint32_t> _m_thread_step    = 0;  // 线程数量调节器 (该值应该不容易变), 线程池会根据任务负载动态增加/减少工作线程数量, 该值指定了工作线程每次增加/减少的数量
    std::atomic<std::int32_t>  _m_thread_stepper = 0;  // 线程数量调节器 (该值正常为 0, 非零时比如 +2 或者 -1, 则管理者线程会增加两个工作线程或者减少一个工作线程以使得该值重新回归 0)

    std::atomic<bool>         _m_managing;    // 管理者线程存活状态
    std::thread               _m_manager;     // 管理者线程
    std::vector<std::thread>  _m_workers;     // 工作线程
    std::set<std::thread::id> _m_exited_ids;  // 工作线程退出后 (未 join) 的 ID 集合


    std::atomic<std::uint32_t>                                                                  _m_queue_max = 0;  // 任务队列最大值
    std::atomic<std::uint32_t>                                                                  _m_queue_cur = 0;  // 任务队列当前任务数量
    std::priority_queue<Task_Item_st, std::vector<Task_Item_st>, Task_Item_st::Task_Comparator> _m_tasks;          // 任务队列

    std::atomic<bool>       _m_auto_threshold;    // 是否自动调整上下限 (根据 CountsOF_MinThreads/CountsOF_MaxThreads)
    std::mutex              _m_mutex_queue;       // 任务队列互斥锁
    std::mutex              _m_mutex_exited_ids;  // 退出线程ID集合互斥锁
    std::mutex              _m_mutex_timing;      // 线程池各种时间互斥锁
    std::condition_variable _m_not_full;          // 条件变量, 表示任务队列有空位(至少存在一个空位)
    std::condition_variable _m_not_empty;         // 条件变量, 表示任务队列有任务(至少存在一个任务)


    std::atomic<std::int64_t>  _m_time_start               = 0;  // 启动的时刻
    std::atomic<std::int64_t>  _m_time_stop                = 0;  // 停止的时刻
    std::atomic<std::int64_t>  _m_time_accumulated_running = 0;  // 累计线程池运行时间
    std::atomic<std::int64_t>  _m_time_accumulated_tasking = 0;  // 累计任务消耗时间
    std::atomic<std::int64_t>  _m_time_avg_task_waiting    = 0;  // 平均任务等待时间
    std::atomic<std::int64_t>  _m_time_avg_task_executing  = 0;  // 平均任务执行时间
    std::atomic<std::uint64_t> _m_counts_accumulated_tasks = 0;  // 累计执行的任务数量
};


#endif
